// Copyright (c) 2021 Presto Labs Pte. Ltd.
// Author: fengyang

#include <filesystem>
#include <iostream>
#include <memory>

#include <cxxopts.hpp>
#include <fmt/format.h>

#include "coin2/app/driver.h"
#include "coin2/base/config/app_config.h"
#include "coin2/base/config/args.h"
#include "coin2/exchange/base/executor/sync_executor.h"
#include "coin2/exchange/base/market/market_exchange_api.h"
#include "coin2/feed/fastfeed/writer.h"
#include "coin2/strategy/strategy.h"

// proto
#include "coin2/exchange/feed.pb.h"

using namespace coin2::base::config;
using namespace coin2::app;
using namespace coin2::exchange::base::symbology;
using namespace coin2::exchange::feed;
using coin2::exchange::base::feed::FeedUpdate;
using fastfeed::proto::KlineInterval;
using fastfeed::proto::KlineInterval_Name;
using coin2::exchange::di::ConstantInjectionBuilder;

static int64_t GetDurationOfInterval(KlineInterval interval) {
  static std::map<KlineInterval, int64_t> interval_map = {
    {fastfeed::proto::KLINE_INTERVAL_1MIN, 60L * 1000000000},
    {fastfeed::proto::KLINE_INTERVAL_3MIN, 3L * 60 * 1000000000},
    {fastfeed::proto::KLINE_INTERVAL_5MIN, 5L * 60 * 1000000000},
    {fastfeed::proto::KLINE_INTERVAL_15MIN, 15L * 60 * 1000000000},
    {fastfeed::proto::KLINE_INTERVAL_30MIN, 30L * 60 * 1000000000},
    {fastfeed::proto::KLINE_INTERVAL_1HOUR, 60L * 60 * 1000000000},
    {fastfeed::proto::KLINE_INTERVAL_1DAY, 24L * 60 * 60 * 1000000000 }
  };
  auto iter = interval_map.find(interval);
  if (iter == interval_map.end()) {
    NOTREACHED() << "Not support interval " << interval;
    return std::numeric_limits<int64_t>::quiet_NaN();
  }

  return iter->second;
}

class KlineCheckerWriter {
 public:
  explicit KlineCheckerWriter(
      const std::string& out_dir,
      int64_t timestamp) {
    if (!out_dir.empty()) {
      auto ptime = PtimeFromTimestamp(timestamp);
      auto YYYYMMDD = boost::gregorian::to_iso_string(ptime.date());
      path_ = fmt::format(
            "{root_dir}/{YYYYMMDD}.csv",
            fmt::arg("root_dir", out_dir),
            fmt::arg("YYYYMMDD", YYYYMMDD));
      if (std::experimental::filesystem::create_directories(path_.parent_path())) {
        LOG(INFO) << "[FileWriter] Created dir: " << path_.parent_path();
      }

      LOG(INFO) << "[FileWriter] Writing to: " << path_;
      of_ = std::make_unique<std::ofstream>(path_);
    }

    const char* header = "mea,symbol,interval,time,error";
    if (of_) {
      *of_ << header << std::endl;
    } else {
      std::cout << header << std::endl;
    }
  }

  std::ostream& Writer() {
    if (of_) {
      return *of_;
    }

    return std::cout;
  }

 private:
  std::experimental::filesystem::path path_;
  std::unique_ptr<std::ofstream> of_;
};

class KlineChecker {
 public:
  explicit KlineChecker(
      const MarketExchangeApi& mea,
      const std::string& symbol,
      const KlineInterval interval,
      std::shared_ptr<KlineCheckerWriter> writer)
      : mea_(mea),
        symbol_(symbol),
        interval_(interval),
        info_writer_(writer) {
  }

  bool Check(const Klines& klines, int64_t timestamp) {
    if (!CheckDuration(klines, timestamp)) {
      return false;
    }
    if (previous_klines_) {
      if (!CheckDiff(*previous_klines_, klines, timestamp)) {
        return false;
      }
    }

    previous_klines_ = klines;
    return true;
  }

 private:
  bool CheckDuration(const Klines& klines, int64_t timestamp) {
    std::optional<Kline> previous_kline;
    for (auto iter = klines.Begin(); iter != klines.End(); iter++) {
      if (previous_kline) {
        if (previous_kline->interval != iter->interval) {
          info_writer_->Writer() << mea_.String() << "," << symbol_ << "," <<
              KlineInterval_Name(interval_) << "," << timestamp << "," <<
              "interval mismatch: " << KlineInterval_Name(previous_kline->interval) <<
              " <=> " << KlineInterval_Name(iter->interval) << std::endl;
          return false;
        }
        int64_t duration = GetDurationOfInterval(iter->interval);
        int64_t diff = previous_kline->open_timestamp - iter->open_timestamp;
        if (duration != diff) {
          info_writer_->Writer() << mea_.String() << "," << symbol_ << "," <<
              KlineInterval_Name(interval_) << "," << timestamp << "," <<
              "duration wrong: " << diff << " <=> " << duration << std::endl;
          return false;
        }
      }

      previous_kline = *iter;
    }
    return true;
  }

  bool CheckDiff(const Klines& previous_klines, const Klines& klines, int64_t timestamp) {
    auto pre_iter = previous_klines.Begin();
    auto iter = klines.Begin();
    while ((pre_iter != previous_klines.End()) && (iter != klines.End())) {
      if (pre_iter->open_timestamp == iter->open_timestamp) {
        if (pre_iter->open != iter->open ||
            pre_iter->close != iter->close ||
            pre_iter->high != iter->high ||
            pre_iter->low != iter->low) {
          info_writer_->Writer() << mea_.String() << "," << symbol_ << "," <<
              KlineInterval_Name(interval_) << "," << timestamp << "," <<
              "kline mismatch: " << iter->open_timestamp << std::endl;
          return false;
        }

        pre_iter++;
        iter++;
      } else if (pre_iter->open_timestamp > iter->open_timestamp) {
        pre_iter++;
      } else if (pre_iter->open_timestamp < iter->open_timestamp) {
        iter++;
      }
    }
    return true;
  }

  std::optional<Klines> previous_klines_;
  MarketExchangeApi mea_;
  std::string symbol_;
  KlineInterval interval_;
  std::shared_ptr<KlineCheckerWriter> info_writer_;
};

class KlineFileWriter {
 public:
  explicit KlineFileWriter(
      const std::string& out_dir,
      const MarketExchangeApi& mea,
      const std::string& symbol,
      KlineInterval interval,
      int64_t timestamp) {
    auto ptime = PtimeFromTimestamp(timestamp);
    auto YYYYMMDD = boost::gregorian::to_iso_string(ptime.date());
    std::string resolution = std::to_string(
        GetDurationOfInterval(interval) / 1000000000);
    path_ = fmt::format(
          "{root_dir}/{YYYYMMDD}/{mea}/{symbol}/{resolution}.csv",
          fmt::arg("root_dir", out_dir),
          fmt::arg("YYYYMMDD", YYYYMMDD),
          fmt::arg("mea", mea.String()),
          fmt::arg("symbol", symbol),
          fmt::arg("resolution", resolution));
    if (std::experimental::filesystem::create_directories(path_.parent_path())) {
      LOG(INFO) << "[FileWriter] Created dir: " << path_.parent_path();
    }

    LOG(INFO) << "[FileWriter] Writing to: " << path_;
    of_ = std::ofstream(path_);
  }

  std::ofstream& Writer() {
    return of_;
  }

 private:
  std::experimental::filesystem::path path_;
  std::ofstream of_;
};

class KlinesWriter {
 public:
  explicit KlinesWriter(const std::string& out_dir, int64_t start_from)
      : out_dir_(out_dir), start_from_(start_from) {}

  void Write(const FeedUpdate& upd) {
    auto& klines = upd.klines();
    auto mea = upd.mea();
    auto symbol = upd.product().absolute_norm();
    auto interval = (*klines.Get())->interval;
    auto mea_s_interval = std::make_tuple(mea, symbol, interval);
    if (klines_writer_.count(mea_s_interval) == 0) {
      klines_writer_[mea_s_interval] = std::make_unique<KlineFileWriter>(
          out_dir_,
          mea,
          symbol,
          interval,
          start_from_);
      klines_writer_[mea_s_interval]->Writer() << "time,open,high,low,close" << std::endl;
    }

    auto& writer = klines_writer_[mea_s_interval]->Writer();
    for (auto iter = klines.RBegin(); iter != klines.REnd(); iter++) {
      if (iter->open_timestamp < start_from_ ||
          iter->open_timestamp <= kline_last_timestamp_[mea_s_interval]) {
        continue;
      }

      writer << Iso8601FromTimestamp(iter->open_timestamp) << ","
              << iter->open << ","
              << iter->high << ","
              << iter->low << ","
              << iter->close << std::endl;
      kline_last_timestamp_[mea_s_interval] = iter->open_timestamp;
    }
  }

 private:
  std::string out_dir_;
  int64_t start_from_;
  std::map<std::tuple<MarketExchangeApi, std::string, KlineInterval>,
           std::unique_ptr<KlineFileWriter>> klines_writer_;
  std::map<std::tuple<MarketExchangeApi, std::string, KlineInterval>,
           int64_t> kline_last_timestamp_;
};

class KlineFeedChecker : public coin2::strategy::IStrategy {
 public:
  explicit KlineFeedChecker(
      const std::string& out_dir,
      const std::string& log_dir)
      : out_dir_(out_dir),
        log_dir_(log_dir){
  }

  void Init(::coin2::app::Driver* driver) override {
    // Warmup period should be handled by parser but ignored by strategy.
    const auto& config = driver->config().feed();
    auto interval = FromIntervalConfig(config.common().archive().interval());
    start_from_ = interval.first;
    if (!out_dir_.empty()) {
      klines_writer_ = std::make_unique<KlinesWriter>(out_dir_, start_from_);
    }

    checker_writer_ = std::make_shared<KlineCheckerWriter>(log_dir_, start_from_);
  }

  void onKlineFeed(const FeedUpdate& upd) override {
    if (upd.timestamp() < start_from_) {
      return;
    }

    auto mea = upd.mea();
    auto symbol = upd.product().absolute_norm();
    auto interval = (*upd.klines().Get())->interval;
    auto mea_s_interval = std::make_tuple(mea, symbol, interval);
    if (klines_checker_.count(mea_s_interval) == 0) {
      klines_checker_[mea_s_interval] = std::make_unique<KlineChecker>(
          mea, symbol, interval, checker_writer_);
    }

    klines_checker_[mea_s_interval]->Check(upd.klines(), upd.timestamp());

    if (klines_writer_) {
      klines_writer_->Write(upd);
    }
  }

 private:
  std::string out_dir_;
  std::string log_dir_;
  int64_t start_from_;
  std::map<std::tuple<MarketExchangeApi, std::string, KlineInterval>,
           std::unique_ptr<KlineChecker>> klines_checker_;
  std::unique_ptr<KlinesWriter> klines_writer_;
  std::shared_ptr<KlineCheckerWriter> checker_writer_;
};

int main(int argc, char* argv[]) {
  DriverConfig app;
  cxxopts::Options opt("Kline Feed Converter", "Generate Kline");
  AddDriverOptions(&opt);
  opt.add_options()
      ("date", "trading date",
          cxxopts::value<std::string>())
      ("out-dir", "kline data export directory",
          cxxopts::value<std::string>()->default_value(""))
      ("log-dir", "checker log export directory",
          cxxopts::value<std::string>()->default_value(""));

  auto res = opt.parse(argc, argv);
  CHECK(ParseDriverOptions(res, &app)) << "Fail to parse driver";

  // port optoin date to dirver config
  if (res.count("date") != 0U) {
    auto* interval = app.mutable_feed()->mutable_common()->mutable_archive()->
        mutable_interval();
    interval->set_start(res["date"].as<std::string>());
  }

  KlineFeedChecker checker(res["out-dir"].as<std::string>(),
                           res["log-dir"].as<std::string>());

  ConstantInjectionBuilder ci_builder = ConstantInjectionBuilder();
  coin2::app::Driver driver(app, &checker, ci_builder);
  driver.Run();

  return 0;
}
